// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: linchuan

#include "coin2/exchange/okex_v5_futures/order_v5/parser.h"

#include <memory>
#include <string>
#include <string_view>
#include <utility>
#include <vector>

#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/okex_v5/order_v5/error_code.h"
#include "coin2/exchange/okex_v5_futures/api_order/native_private_client.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::okex_v5_futures::order_v5 {

namespace {

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::CurrencyTransferElement;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;
using coin::proto::TransferRequestProto;
using coin2::exchange::base::order::IOrderContext;
using coin2::exchange::okex::symbology::OkexCurrency;
using coin2::exchange::okex::symbology::OkexProduct;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;
using TransferType = coin::proto::TransferRequestProto::TransferType;
using OkexV5FuturesOrderSubTopicId =
    coin2::exchange::okex_v5_futures::api_order::OkexV5FuturesOrderSubTopicId;

inline coin::proto::OrderSide GetOrderSide(const JsonValue& data) {
  using coin::proto::OrderSide;
  std::string_view direction = data["side"].GetString();
  if (direction == "buy") {
    return OrderSide::BUY_ORDER;
  } else if (direction == "sell") {
    return OrderSide::SELL_ORDER;
  } else {
    return OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline int64_t GetProcOrderId(const JsonValue& data) {
  int64_t proc_order_id = 0;
  if (data.HasMember("clOrdId")) {
    std::string_view proc_str = data["clOrdId"].GetString();
    try {
      proc_order_id = stoll(proc_str.substr(2));
    } catch (std::exception& e) {
      LOG(ERROR) << e.what() << JsonToString(data);
    }
  }
  return proc_order_id;
}

bool IsOkexErrMessage(const JsonValue& doc) {
  if (doc.HasMember("code") && std::string(doc["code"].GetString()) != "0") {
    return true;
  }
  if (doc.HasMember("msg") && std::string(doc["msg"].GetString()) != "") {
    return true;
  }
  return false;
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (!(json.HasMember("arg") && json["arg"].HasMember("channel") &&
        json["arg"].HasMember("instType"))) {
    return false;
  }

  return std::string(json["arg"]["channel"].GetString()) == "orders";
}

bool IsPositionUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (!(json.HasMember("arg") && json["arg"].HasMember("channel") &&
        json["arg"].HasMember("instType"))) {
    return false;
  }

  return std::string(json["arg"]["channel"].GetString()) == "positions";
}

inline double stod(std::string_view s) {
  if (s == "") {
    return 0;
  }
  return ::stod(s.data());
}
inline double stod(const char* s) {
  if (s == nullptr) {
    return 0;
  }
  return stod(std::string_view(s));
}

}  // namespace

void OkexV5FuturesParser::ParseRestAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_balance = ParseBalance(document);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountBalance OkexV5FuturesParser::ParseBalance(const JsonValue& doc) {
  RawBalanceMap raw_balance_map;
  auto success = ParseRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }

  AccountBalance acc_balance;
  acc_balance.set_exchange("Okex");
  acc_balance.set_market_type("Futures");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }
  is_account_balance_initialized_ = true;

  return acc_balance;
}

bool OkexV5FuturesParser::ParseRawBalance(
    const JsonValue& document,
    RawBalanceMap* raw_balance_map) {
  // Account balance
  /*
  {
      "code": "0",
      "msg": "",
      "data": [{
          "uTime": "1597026383085",
          "totalEq": "41624.32",
          "isoEq": "3624.32",
          "adjEq": "41624.32",
          "imr": "4162.33",
          "mmr": "4",
          "mgnRatio": "41624.32",
          "details": [{
                  "ccy": "BTC",
                  "eq": "123",
                  "isoEq":"",
                  "availEq": "1234",
                  "availBal": "",
                  "frozenBal": "14",
                  "upl": "124",
                  "mgnRatio": "",
                  "interest": "12",
                  "liab": "1"
              },
              ...
          ]
      }]
  }
  */
  // Asset balance
  /*
  {
      "code": "0",
      "msg": "",
      "data": [{
              "availBal": "37.11827078",
              "bal": "37.11827078",
              "ccy": "ETH",
              "frozenBal": "0"
          }
      ]
  }
  */
  if (document.IsObject() && IsOkexErrMessage(document)) {
    LOG(ERROR) << "[ParseRawBalance] " << JsonToString(document);
    return false;
  }
  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRawBalance] " << JsonToString(document);
    return false;
  }

  if (document["data"].GetArray().Size() > 0) {
    if (document["data"][0].HasMember("details")) {
      for (const auto& data : document["data"].GetArray()) {
        for (const auto& detail : data["details"].GetArray()) {
          if (!detail.HasMember("ccy") || !detail.HasMember("eq") || !detail.HasMember("availEq")) {
            continue;
          }
          CurrencyBalance balance;
          balance.set_currency_native(detail["ccy"].GetString());
          auto total = stod(detail["eq"].GetString());
          auto avail = stod(detail["availEq"].GetString());
          balance.set_total(total);
          balance.set_hold(total - avail);
          balance.set_available(avail);
          raw_balance_map->emplace(balance.currency_native(), balance);
        }
      }
    } else if (document["data"][0].HasMember("ccy")) {
      for (const auto& data : document["data"].GetArray()) {
        if (!data.HasMember("ccy") || !data.HasMember("bal") || !data.HasMember("availBal")) {
          continue;
        }
        CurrencyBalance balance;
        balance.set_currency_native(data["ccy"].GetString());
        auto total = stod(data["bal"].GetString());
        auto avail = stod(data["availBal"].GetString());
        balance.set_total(total);
        balance.set_hold(total - avail);
        balance.set_available(avail);
        raw_balance_map->emplace(balance.currency_native(), balance);
      }
    }
  }

  return true;
}

void OkexV5FuturesParser::ParseRestAccountPosition() {
  JsonDocument doc;
  if (!PrepareJsonDocument(&doc)) {
    LOG(ERROR) << "[ParseRestAccountPosition] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_position = ParsePosition(doc);
  if (!account_position.has_exchange() || !account_position.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  result_.set_type(MsgType::ACCOUNT_POSITION);
  *result_.mutable_account_position() = std::move(account_position);
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountPosition OkexV5FuturesParser::ParsePosition(const JsonValue& doc) {
  RawPositionMap raw_position_map;
  auto success = ParseRawPosition(doc, &raw_position_map);
  if (!success) {
    return {};
  }

  AccountPosition acc_position;
  acc_position.set_exchange("Okex");
  acc_position.set_market_type("Futures");
  for (const auto& native_symbol : symcache_->GetProductNative()) {
    if (raw_position_map.count(native_symbol) != 0) {
      continue;
    }
    auto* position = acc_position.add_each_position();
    position->set_long_position(0);
    position->set_short_position(0);
    position->set_net_position(0);
    position->set_available_long_position(0);
    position->set_available_short_position(0);
    position->set_native_symbol(native_symbol);

    DLOG(INFO) << "POS: " << native_symbol << " " << position->net_position();
  }

  for (auto& pair : raw_position_map) {
    auto* position = acc_position.add_each_position();
    *position = std::move(pair.second);
    position->set_native_symbol(pair.first);
    DLOG(INFO) << "POS: " << pair.first << " " << position->net_position();
  }
  is_account_position_initialized_ = true;

  return acc_position;
}

bool OkexV5FuturesParser::ParseRawPosition(
    const JsonValue& document,
    RawPositionMap* raw_position_map) {
  /*
  {
    "code": "0",
    "msg": "",
    "data": [{
        "instId": "BTC-USDT",
        "instType": "MARGIN",
        "mgnMode": "cross",
        "posId": "111111222",
        "posSide": "net",
        "pos": "10",
        "ccy": "BTC",
        "posCcy": "BTC",
        "availPos": "2",
        "avgPx": "3320",
        "upl": "0.00020928",
        "uplRatio": "0.00020928",
        "lever": "2",
        "liqPx": "0.00020928",
        "imr": "2",
        "margin": "",
        "mgnRatio": "",
        "mmr": "2",
        "liab": "0.00020928",
        "liabCcy": "USDT",
        "interest": "2",
        "tradeId": "2",
        "optVal": "",
        "adl": "2",
        "last":"12353.5",
        "cTime": "1597026383085",
        "uTime": "1597026383085"
    }]
  }
  */
  if (document.IsObject() && IsOkexErrMessage(document)) {
    LOG(ERROR) << "[ParsePosition] " << JsonToString(document);
    return false;
  }
  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParsePosition] " << JsonToString(document);
    return false;
  }

  for (const auto& data : document["data"].GetArray()) {
    if (!data.HasMember("instId") || !data.HasMember("pos") || !data.HasMember("posSide") ||
        !data.HasMember("availPos") || !data.HasMember("mgnMode")) {
      LOG(ERROR) << "[ParsePosition] Parsing Error. " << JsonToString(data);
      continue;
    }
    std::string_view side(data["posSide"].GetString());
    if (!used_in_python_) {
      if (!CheckCrossModeConsistent(data, "mgnMode")) {
        LOG(INFO) << JsonToString(data);
        continue;
      }
    }

    std::string native_symbol = data["instId"].GetString();
    auto& position = (*raw_position_map)[native_symbol];

    if (side == "long") {
      double long_position = stod(data["pos"].GetString());
      double long_avail_position = stod(data["availPos"].GetString());
      position.set_long_position(long_position);
      position.set_available_long_position(long_avail_position);
    } else if (side == "short") {
      double short_position = -stod(data["pos"].GetString());
      double short_avail_position = -stod(data["availPos"].GetString());
      position.set_short_position(short_position);
      position.set_available_short_position(short_avail_position);
    } else if (side == "net") {
      double net_position = stod(data["pos"].GetString());
      if (net_position >= 0) {
        position.set_long_position(net_position);
        position.set_available_long_position(net_position);
        position.set_short_position(0.);
        position.set_available_short_position(0.);
      } else {
        position.set_short_position(net_position);
        position.set_available_short_position(net_position);
        position.set_long_position(0.);
        position.set_available_long_position(0.);
      }
    } else {
      CHECK(false) << JsonToString(data);
    }
    position.set_net_position(position.long_position() + position.short_position());
  }

  return true;
}

void OkexV5FuturesParser::ParseWsPositionInfoList(const JsonDocument& doc) {
  auto account_position = ParsePositionFromWs(doc);
  if (!account_position.has_exchange() || !account_position.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  result_.set_type(MsgType::ACCOUNT_POSITION);
  *result_.mutable_account_position() = std::move(account_position);
}

AccountPosition OkexV5FuturesParser::ParsePositionFromWs(const JsonValue& doc) {
  RawPositionMap raw_position_map;
  bool success = ParseRawPositionFromWs(doc, &raw_position_map);
  if (!success) {
    return {};
  }

  AccountPosition acc_position;
  acc_position.set_exchange("Okex");
  acc_position.set_market_type("Futures");
  for (auto& pair : raw_position_map) {
    auto* position = acc_position.add_each_position();
    *position = std::move(pair.second);
    position->set_native_symbol(pair.first);
    DLOG(INFO) << "POS: " << pair.first << " " << position->net_position();
  }

  return acc_position;
}

bool OkexV5FuturesParser::ParseRawPositionFromWs(
    const JsonValue& document,
    RawPositionMap* raw_position_map) {
  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseWsPosition] " << JsonToString(document);
    return false;
  }

  for (const auto& data : document["data"].GetArray()) {
    if (!data.HasMember("instId") || !data.HasMember("pos") || !data.HasMember("posSide") ||
        !data.HasMember("availPos") || !data.HasMember("mgnMode")) {
      LOG(ERROR) << "[ParsePosition] Parsing Error. " << JsonToString(data);
      continue;
    }
    std::string_view side(data["posSide"].GetString());
    if (!CheckCrossModeConsistent(data, "mgnMode")) {
      LOG(INFO) << JsonToString(data);
      continue;
    }

    std::string native_symbol = data["instId"].GetString();
    auto& position = (*raw_position_map)[native_symbol];

    if (side == "long") {
      double long_position = stod(data["pos"].GetString());
      double long_avail_position = stod(data["availPos"].GetString());
      position.set_long_position(long_position);
      position.set_available_long_position(long_avail_position);
    } else if (side == "short") {
      double short_position = -stod(data["pos"].GetString());
      double short_avail_position = -stod(data["availPos"].GetString());
      position.set_short_position(short_position);
      position.set_available_short_position(short_avail_position);
    } else if (side == "net") {
      double net_position = stod(data["pos"].GetString());
      if (net_position >= 0) {
        position.set_long_position(net_position);
        position.set_available_long_position(net_position);
        position.set_short_position(0.);
        position.set_available_short_position(0.);
      } else {
        position.set_short_position(net_position);
        position.set_available_short_position(net_position);
        position.set_long_position(0.);
        position.set_available_long_position(0.);
      }
    } else {
      CHECK(false) << JsonToString(data);
    }
    position.set_net_position(position.long_position() + position.short_position());
  }

  return true;
}

void OkexV5FuturesParser::ParseOrderInfo(
    const JsonValue& data,
    coin::proto::ProductOrderElement* order) {
  order->set_market_type(MarketType::Futures);
  order->set_exchange_type(ExchangeType::Okex);
  order->set_api_version("v5");
  order->set_exchange_order_id(data["ordId"].GetString());
  order->set_symbol(data["instId"].GetString());
  order->set_price(stod(data["px"].GetString()));
  order->set_qty(stod(data["sz"].GetString()));
  order->set_proc_order_id(GetProcOrderId(data));
  order->set_side(GetOrderSide(data));
  std::string_view type_str = data["ordType"].GetString();
  if (type_str == "market") {
    order->set_order_type(coin::proto::OrderType::MARKET_ORDER);
  } else {
    order->set_order_type(coin::proto::OrderType::LIMIT_ORDER);
  }
  order->set_timestamp(msg_->timestamp);
  const std::string state = data["state"].GetString();
  if (state == "live" || state == "partially_filled") {
    order->set_is_live(true);
  } else if (state == "canceled") {
    order->set_cancel_confirmed(true);
  } else if (state == "filled") {
    order->set_fully_filled(true);
  } else {
    LOG(ERROR) << "Unknown order status! " << state;
  }
}

void OkexV5FuturesParser::ParseRestOpenOrders() {
  /*
  {
    "code":"0",
    "msg":"",
    "data":[
        {
            "instType":"FUTURES",
            "instId":"BTC-USD-200329",
            "ccy":"",
            "ordId":"123445",
            "clOrdId":"b1",
            "tag":"",
            "px":"999",
            "sz":"3",
            "pnl":"5",
            "ordType":"limit",
            "side":"buy",
            "posSide":"long",
            "tdMode":"isolated",
            "accFillSz":"isolated",
            "fillPx":"0",
            "tradeId":"0",
            "fillSz":"0",
            "fillTime":"0",
            "state":"live",
            "avgPx":"0",
            "lever":"20",
            "tpTriggerPx":"",
            "tpOrdPx":"",
            "slTriggerPx":"",
            "slOrdPx":"",
            "feeCcy":"",
            "fee":"",
            "rebateCcy":"",
            "rebate":"",
            "category":"",
            "uTime":"1597026383085",
            "cTime":"1597026383085"
        }
    ]
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();

  if (document.IsObject() && IsOkexErrMessage(document)) {
    LOG(ERROR) << "[ParseRestOrders] " << JsonToString(document);
    return;
  }
  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestOrders] " << JsonToString(document);
    return;
  }

  int record_cnt = 0;
  std::string after_order_id = "";
  bool product_order_bundle_found = false;
  for (const auto& info : document["data"].GetArray()) {
    if (!CheckCrossModeConsistent(info, "tdMode")) {
      continue;
    }

    if (std::string_view(info["ordType"].GetString()) == "market") {
      // Ignore market order
      continue;
    }

    auto order = product_order_bundle->add_each_order();
    ParseOrderInfo(info, order);
    product_order_bundle_found = true;
    ++record_cnt;
    after_order_id = order->exchange_order_id();
  }

  if (record_cnt >= 100) {
    mb_->PublishEvent("okex_futures/query_open_orders", &after_order_id);
  }

  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

void OkexV5FuturesParser::ParseRestHistOrders() {
  /*
  {
    "code":"0",
    "msg":"",
    "data":[
        {
            "instType":"FUTURES",
            "instId":"BTC-USD-200329",
            "ccy":"",
            "ordId":"123445",
            "clOrdId":"b1",
            "tag":"",
            "px":"999",
            "sz":"3",
            "pnl":"5",
            "ordType":"limit",
            "side":"buy",
            "posSide":"long",
            "tdMode":"isolated",
            "accFillSz":"isolated",
            "fillPx":"0",
            "tradeId":"0",
            "fillSz":"0",
            "fillTime":"0",
            "state":"live",
            "avgPx":"0",
            "lever":"20",
            "tpTriggerPx":"",
            "tpOrdPx":"",
            "slTriggerPx":"",
            "slOrdPx":"",
            "feeCcy":"",
            "fee":"",
            "rebateCcy":"",
            "rebate":"",
            "category":"",
            "uTime":"1597026383085",
            "cTime":"1597026383085"
        }
    ]
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();

  if (document.IsObject() && IsOkexErrMessage(document)) {
    LOG(ERROR) << "[ParseRestOrders] " << JsonToString(document);
    return;
  }
  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestOrders] " << JsonToString(document);
    return;
  }

  bool product_order_bundle_found = false;
  for (const auto& info : document["data"].GetArray()) {
    if (!CheckCrossModeConsistent(info, "tdMode")) {
      continue;
    }

    if (std::string_view(info["ordType"].GetString()) == "market") {
      // Ignore market order
      continue;
    }

    auto order = product_order_bundle->add_each_order();
    ParseOrderInfo(info, order);
    product_order_bundle_found = true;
  }

  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

void OkexV5FuturesParser::ParseWsOrderInfoList(const JsonDocument& document) {
  /*
  {
    "arg": {
        "channel": "orders",
        "instType": "FUTURES",
        "instId": "BTC-USD-200329"
    },
    "data": [{
        "instType": "FUTURES",
        "instId": "BTC-USD-200329",
        "ordId": "123445",
        "clOrdId": "b1",
        "tag": "",
        "px": "999",
        "sz": "3",
        "ordType": "limit",
        "side": "buy",
        "posSide": "long",
        "tdMode": "cross",
        "fillSz": "0",
        "fillPx": "long",
        "tradeId": "0",
        "accFillSz": "323",
        "fillTime": "0",
        "state": "canceled",
        "avgPx": "0",
        "lever": "20",
        "tpTriggerPx": "0",
        "tpOrdPx": "20",
        "slTriggerPx": "0",
        "slOrdPx": "20",
        "feeCcy": "",
        "fee": "",
        "rebateCcy": "",
        "rebate": "",
        "pnl": "",
        "category": "",
        "uTime": "1597026383085",
        "cTime": "1597026383085",
        "reqId": "",
        "amendResult": "",
        "code": "0",
        "msg": ""
    }]
  }
  */
  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();

  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseWsOrderInfoList] " << JsonToString(document);
    return;
  }

  bool product_order_bundle_found = false;
  for (const auto& info : document["data"].GetArray()) {
    if (IsOkexErrMessage(info)) {
      LOG(ERROR) << "[ParseWsOrderInfoList] " << JsonToString(info);
      continue;
    }
    if (!CheckCrossModeConsistent(info, "tdMode")) {
      continue;
    }
    auto order = product_order_bundle->add_each_order();
    ParseOrderInfo(info, order);
    product_order_bundle_found = true;
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

void OkexV5FuturesParser::ParseFillInfo(
    const JsonValue& data,
    coin::proto::ProductFillElement* fill) {
  fill->set_symbol(data["instId"].GetString());
  fill->set_exchange_order_id(data["ordId"].GetString());
  fill->set_price(stod(data["fillPx"].GetString()));
  fill->set_qty(stod(data["fillSz"].GetString()));
  fill->set_fill_id(data["tradeId"].GetString());
  if (data.HasMember("fillFee")) {
    if (std::string fee = data["fillFee"].GetString(); !fee.empty()) {
      fill->set_fee(-stod(fee));
    }
  }
  if (data.HasMember("fillFeeCcy")) {
    if (std::string fee_currency = data["fillFeeCcy"].GetString(); !fee_currency.empty()) {
      fill->set_fee_currency(fee_currency);
    }
  }
  if (!data.HasMember("execType")) {
    // Infer fill type logic, for ws msg
    auto oc = ocm_->FindByProcOrderId(fill->proc_order_id());
    if (oc) {
      if (std::abs(fill->price() - oc->order_spec().price) > 1e-6) {
        fill->set_fill_type(FillType::TAKER_FILL_TYPE);
      } else if (oc->order_spec().post_only) {
        fill->set_fill_type(FillType::MAKER_FILL_TYPE);
      } else if (fill->fill_timestamp() - oc->order_created_time() > 5e9) {
        fill->set_fill_type(FillType::MAKER_FILL_TYPE);
      } else {
        fill->set_fill_type(FillType::TAKER_FILL_TYPE);
      }
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    return;
  }

  std::string_view exec_type_str = data["execType"].GetString();
  if (exec_type_str == "M") {
    fill->set_fill_type(FillType::MAKER_FILL_TYPE);
  } else if (exec_type_str == "T") {
    fill->set_fill_type(FillType::TAKER_FILL_TYPE);
  } else {
    fill->set_fill_type(FillType::UNKNOWN_FILL_TYPE);
  }
}

void OkexV5FuturesParser::ParseRestFillOrders() {
  /*
  {
    "code":"0",
    "msg":"",
    "data":[
         {
            "instType":"FUTURES",
            "instId":"BTC-USD-200329",
            "tradeId":"123",
            "ordId":"123445",
            "billId":"1111",
            "tag":"",
            "fillPx":"999",
            "fillSz":"3",
            "side":"buy",
            "posSide":"long",
            "execType":"M",
            "feeCcy":"",
            "fee":"",
            "ts":"1597026383085"
        },
        ...
    ]
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestFillOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();

  if (document.IsObject() && IsOkexErrMessage(document)) {
    LOG(ERROR) << "[ParseRestFillOrders] " << JsonToString(document);
    return;
  }

  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestFillOrders] " << JsonToString(document);
    return;
  }

  bool product_fill_bundle_found = false;
  for (const auto& info : document["data"].GetArray()) {
    auto fill = product_fill_bundle->add_each_fill();
    ParseFillInfo(info, fill);
    product_fill_bundle_found = true;
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

void OkexV5FuturesParser::ParseWsFillInfoList(const JsonDocument& document) {
  // same as ParseWsOrderInfoList
  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();

  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseWsFillInfoList] " << JsonToString(document);
    return;
  }

  bool product_fill_bundle_found = false;
  for (const auto& info : document["data"].GetArray()) {
    if (IsOkexErrMessage(info)) {
      LOG(ERROR) << "[ParseWsFillInfoList] " << JsonToString(info);
      continue;
    }
    const std::string fill_sz = info["fillSz"].GetString();
    if (fill_sz == "" || fill_sz == "0") {
      continue;
    }
    auto fill = product_fill_bundle->add_each_fill();
    ParseFillInfo(info, fill);
    product_fill_bundle_found = true;
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

void OkexV5FuturesParser::ParseSubmitResponse() {
  /*
  {
      "code":"0",
      "msg":"",
      "data":[
          {
              "clOrdId":"oktswap6",
              "ordId":"12345689",
              "tag":"",
              "sCode":"0",
              "sMsg":""
          }
      ]
  }
  */
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[OkexFuturesSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[OkexFuturesSubmitOrder] Parsing Error." << msg_->DebugString();
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (!document.HasMember("code") || !document.HasMember("msg") || !document.HasMember("data") ||
      !document["data"].IsArray() || document["data"].Size() != 1 ||
      !document["data"][0].HasMember("sCode") || !document["data"][0].HasMember("sMsg") ||
      !document["data"][0].HasMember("ordId")) {
    LOG(ERROR) << "[OkexFuturesSubmitOrder] Parsing Error." << msg_->DebugString();
    const auto http_status_code = http_context->res.result_int();
    const std::string error_code = document["code"].GetString();
    if (400 == http_status_code && error_code == "50004") {
      // does not mean that the request was successful or failed, please check the request result
      result_.Clear();
      result_.set_type(MsgType::IGNORABLE);
    } else {
      resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
      resp->set_success(false);
    }
    return;
  }
  auto error_code = GetOrderErrorCode(document["data"][0]["sCode"].GetString());
  resp->set_error_code(error_code);
  resp->set_exchange_error_code(document["data"][0]["sCode"].GetString());
  resp->set_exchange_error_msg(document["data"][0]["sMsg"].GetString());
  if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
    LOG(ERROR) << JsonToString(document);
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_exchange_order_id(document["data"][0]["ordId"].GetString());
}

void OkexV5FuturesParser::ParseBatchSubmitResponse() {
  /*
  {
      "code":"0",
      "msg":"",
      "data":[
          {
              "clOrdId":"oktswap6",
              "ordId":"12345689",
              "tag":"",
              "sCode":"0",
              "sMsg":""
          },
          {
              "clOrdId":"oktswap7",
              "ordId":"12344",
              "tag":"",
              "sCode":"0",
              "sMsg":""
          }
      ]
  }
  */
  const auto& oc_list = msg_->oc_list;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::MULTI_SUBMIT_RESPONSE);
  auto* response = result_.mutable_multi_submit_response();
  response->Clear();

  std::unordered_map<int64_t, SubmitOrderResponse*> proc_order_id_to_response_map;
  for (const auto* order_context : oc_list) {
    auto* resp = response->add_each_submit_response();

    resp->set_symbol(order_context->order_spec().product);
    resp->set_proc_order_id(order_context->proc_order_id());
    resp->set_http_status(http_context->res.result_int());
    resp->set_timestamp(msg_->timestamp);
    resp->set_success(false);

    CHECK(proc_order_id_to_response_map.count(order_context->proc_order_id()) == 0)
        << "proc_order_id duplicate proc_order_id:" << order_context->proc_order_id();
    proc_order_id_to_response_map[order_context->proc_order_id()] = resp;
  }

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[" << __func__ << "] AsioHttpContextStatus Error";
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    }
    return;
  }
  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[" << __func__ << "] API_LIMIT_ERROR";
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    }
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[" << __func__ << "] Parse Error\n" << http_context->res;
    for (auto& resp : *(response->mutable_each_submit_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    }
    return;
  }

  if (!json.IsObject() || !json.HasMember("code") || !json.HasMember("msg") ||
      !json.HasMember("data") || !json["data"].IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Parse Error\n" << http_context->res;
    const auto http_status_code = http_context->res.result_int();
    const std::string error_code = json["code"].GetString();
    if (400 == http_status_code && error_code == "50004") {
      // does not mean that the request was successful or failed, please check the request result
      result_.Clear();
      result_.set_type(MsgType::IGNORABLE);
    } else {
      for (auto& resp : *(response->mutable_each_submit_response())) {
        resp.set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
      }
    }
    return;
  }

  const auto& array = json["data"].GetArray();

  CHECK_EQ(proc_order_id_to_response_map.size(), array.Size())
      << "Batch submit response size not equal to request size response:" << JsonToString(json)
      << " request:" << response->DebugString();

  for (const auto& data : array) {
    int64_t client_order_id = GetProcOrderId(data);

    CHECK(proc_order_id_to_response_map.count(client_order_id) != 0)
        << "unknown client order id found client_order_id:" << client_order_id;

    auto* resp = proc_order_id_to_response_map[client_order_id];

    auto error_code = GetOrderErrorCode(data["sCode"].GetString());
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(data["sCode"].GetString());
    resp->set_exchange_error_msg(data["sMsg"].GetString());
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << JsonToString(data);
      continue;
    }

    resp->set_success(true);
    resp->set_exchange_order_id(data["ordId"].GetString());
  }
}

void OkexV5FuturesParser::ParseCancelResponse() {
  // same as ParseSubmitResponse
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[OkexFuturesCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[OkexFuturesCancelOrder] Parsing Error." << msg_->DebugString();
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (!document.HasMember("code") || !document.HasMember("msg") || !document.HasMember("data") ||
      !document["data"].IsArray() || document["data"].Size() != 1 ||
      !document["data"][0].HasMember("sCode") || !document["data"][0].HasMember("sMsg") ||
      !document["data"][0].HasMember("ordId")) {
    LOG(ERROR) << "[OkexFuturesCancelOrder] Parsing Error." << msg_->DebugString();
    const auto http_status_code = http_context->res.result_int();
    const std::string error_code = document["code"].GetString();
    if (400 == http_status_code && error_code == "50004") {
      // does not mean that the request was successful or failed, please check the request result
      result_.Clear();
      result_.set_type(MsgType::IGNORABLE);
    } else {
      resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
      resp->set_success(false);
    }
    return;
  }

  auto error_code = GetOrderErrorCode(document["data"][0]["sCode"].GetString());
  resp->set_error_code(error_code);
  resp->set_exchange_error_code(document["data"][0]["sCode"].GetString());
  resp->set_exchange_error_msg(document["data"][0]["sMsg"].GetString());
  if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
    LOG(ERROR) << "Cancel failed! " << JsonToString(document) << ", "
               << order_context->exchange_order_id();
    resp->set_success(false);
    return;
  }

  resp->set_exchange_order_id(std::string(document["data"][0]["ordId"].GetString()));
  resp->set_success(true);
}

void OkexV5FuturesParser::ParseBatchCancelResponse() {
  /*
  {
      "code":"0",
      "msg":"",
      "data":[
          {
              "clOrdId":"oktswap6",
              "ordId":"12345689",
              "sCode":"0",
              "sMsg":""
          },
          {
              "clOrdId":"oktswap7",
              "ordId":"12344",
              "sCode":"0",
              "sMsg":""
          }
      ]
  }
  */

  const auto& oc_list = msg_->oc_list;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::MULTI_CANCEL_RESPONSE);
  auto* response = result_.mutable_multi_cancel_response();
  response->Clear();

  std::unordered_map<int64_t, CancelOrderResponse*> proc_order_id_to_response_map;
  std::unordered_map<std::string, CancelOrderResponse*> order_id_to_response_map;
  for (const auto* order_context : oc_list) {
    auto* resp = response->add_each_cancel_response();

    resp->set_symbol(order_context->order_spec().product);
    resp->set_proc_order_id(order_context->proc_order_id());
    resp->set_http_status(http_context->res.result_int());
    resp->set_exchange_order_id(order_context->exchange_order_id());
    resp->set_timestamp(msg_->timestamp);
    resp->set_success(false);

    CHECK(proc_order_id_to_response_map.count(order_context->proc_order_id()) == 0)
        << "proc_order_id duplicate proc_order_id:" << order_context->proc_order_id();
    proc_order_id_to_response_map[order_context->proc_order_id()] = resp;

    if (order_context->has_exchange_order_id()) {
      CHECK(order_id_to_response_map.count(order_context->exchange_order_id()) == 0)
          << "exchange_order_id duplicate exchange_order_id:" << order_context->exchange_order_id();
      order_id_to_response_map[order_context->exchange_order_id()] = resp;
    }
  }

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[" << __func__ << "] AsioHttpContextStatus Error";
    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    }
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[" << __func__ << "] API_LIMIT_ERROR";
    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    }
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json) || !json.IsObject() || !json.HasMember("data") ||
      !json["data"].IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Parse Error\n" << http_context->res;
    if (json.IsObject() && json.HasMember("code") && json["code"].IsString()) {
      const auto http_status_code = http_context->res.result_int();
      const std::string error_code = json["code"].GetString();
      if (400 == http_status_code && error_code == "50004") {
        // does not mean that the request was successful or failed, please check the request result
        result_.Clear();
        result_.set_type(MsgType::IGNORABLE);
        return;
      }
    }

    for (auto& resp : *(response->mutable_each_cancel_response())) {
      resp.set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    }
    return;
  }

  const auto& array = json["data"].GetArray();

  CHECK_EQ(proc_order_id_to_response_map.size(), array.Size())
      << "Batch cancel response size not equal to request size msg:" << msg_->DebugString();

  for (const auto& data : array) {
    int64_t client_order_id = GetProcOrderId(data);
    std::string order_id = data["ordId"].GetString();

    CancelOrderResponse* resp;
    if (proc_order_id_to_response_map.count(client_order_id) != 0) {
      resp = proc_order_id_to_response_map[client_order_id];
    } else if (order_id_to_response_map.count(order_id) != 0) {
      resp = order_id_to_response_map[order_id];
    } else {
      NOTREACHED() << "response: " << JsonToString(data)
                   << ", can not find the CancelOrderResponse for this cancel response!";
    }

    auto error_code = GetOrderErrorCode(data["sCode"].GetString());
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(data["sCode"].GetString());
    resp->set_exchange_error_msg(data["sMsg"].GetString());
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "Cancel failed! " << JsonToString(data) << ", " << resp->exchange_order_id();
      continue;
    }

    resp->set_success(true);
    resp->set_exchange_order_id(order_id);
  }
}

void OkexV5FuturesParser::ParseRestDeposit() {
  /*
  {
    "code": "0",
    "data": [{
      "amt": "100000",
      "ccy": "USDT",
      "chain": "USDT-TRC20",
      "depId": "30556253",
      "from": "",
      "state": "2",
      "to": "TX5krATmLzJ69iJB9tgf7uVWoJiisMuWJR",
      "ts": "1635254454000",
      "txId": "a35d536f637729f20ffe740d0324b7dfd156d38ee43d418b04f80d14b42adc5b"
    }],
    "msg": ""
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestDeposit] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsOkexErrMessage(document)) {
    LOG(ERROR) << "[ParseRestDeposit] Error message\n" << JsonToString(document);
    return;
  }

  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestDeposit] Invalid data format\n" << JsonToString(document);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& data : document["data"].GetArray()) {
    if (!data.HasMember("ccy") || !data.HasMember("amt")) {
      LOG(ERROR) << "[ParseRestDeposit] Invalid transfer format";
      continue;
    }
    if (!symcache_->has_currency_native(data["ccy"].GetString())) {
      continue;
    }
    std::string transfer_id = data["depId"].GetString();
    if (transfer_deposit_set_.count(transfer_id) > 0) {
      continue;
    }
    transfer_deposit_set_.insert(transfer_id);
    auto* currency = symcache_->GetCurrencyByNative(data["ccy"].GetString());
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_type(TransferRequestProto::DEPOSIT);
    std::string state = data["state"].GetString();
    if ((state == "0") || (state == "8")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_SUBMITTED);
    } else if (state == "1") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_ACCEPTED);
    } else if (state == "2") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    }
    currency_transfer->set_transfer_id(transfer_id);
    currency_transfer->set_qty(stod(data["amt"].GetString()));
    currency_transfer->set_from_address(data["from"].GetString());
    currency_transfer->set_to_address(data["to"].GetString());
    currency_transfer->set_create_timestamp(stoll(data["ts"].GetString()) * 1e6);
    currency_transfer_bundle_found = true;
  }

  if (!is_transfer_deposit_initialized_) {
    is_transfer_deposit_initialized_ = true;
    LOG(INFO) << "[ParseRestDeposit] Deposit ignored: " << transfer_deposit_set_.size();
    return;
  }

  if (currency_transfer_bundle_found) {
    LOG(INFO) << "[ParseRestDeposit] Deposit parsed: "
              << currency_transfer_bundle->each_transfer_size();
    result_.set_type(MsgType::TRANSFER);
  }
}

void OkexV5FuturesParser::ParseRestWithdrawal() {
  /*
  {
    "code": "0",
    "data": [{
      "chain": "BTC-Bitcoin",
      "fee": "0.0004",
      "ccy": "BTC",
      "amt": "4",
      "txId": "929b5f91beb710b4e1974fec5e1e031b3ff2d9e5b3bd1ed4baab1f810ed029aa",
      "from": "87338117",
      "to": "1NtxcSC6XfpUXkbUbCHqa15UYv4nzJUNTL",
      "state": "2",
      "ts": "1621838350000",
      "wdId": "15200565"
    }],
    "msg": ""
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestWithdrawal] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsOkexErrMessage(document)) {
    LOG(ERROR) << "[ParseRestWithdrawal] Error message\n" << JsonToString(document);
    return;
  }

  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[ParseRestWithdrawal] Invalid data format\n" << JsonToString(document);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& data : document["data"].GetArray()) {
    if (!data.HasMember("ccy") || !data.HasMember("amt")) {
      LOG(ERROR) << "[ParseRestWithdrawal] Invalid transfer format";
      continue;
    }
    if (!symcache_->has_currency_native(data["ccy"].GetString())) {
      continue;
    }
    std::string transfer_id = data["wdId"].GetString();
    if (transfer_withdrawal_set_.count(transfer_id) > 0) {
      continue;
    }
    transfer_withdrawal_set_.insert(transfer_id);
    auto* currency = symcache_->GetCurrencyByNative(data["ccy"].GetString());
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_type(TransferRequestProto::WITHDRAW);
    std::string state = data["state"].GetString();
    if (state == "-3") {
      currency_transfer->set_status(TransferRequestProto::CANCEL_SUBMITTED);
    } else if (state == "-2") {
      currency_transfer->set_status(TransferRequestProto::CANCEL_CONFIRMED);
    } else if (state == "-1") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_REJECTED);
    } else if ((state == "0") || (state == "1")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_SUBMITTED);
    } else if (state == "2") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    } else if ((state == "3") || (state == "4") || (state == "5")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_ACCEPTED);
    }
    currency_transfer->set_transfer_id(data["wdId"].GetString());
    currency_transfer->set_qty(stod(data["amt"].GetString()));
    currency_transfer->set_from_address(data["from"].GetString());
    currency_transfer->set_to_address(data["to"].GetString());
    currency_transfer->set_create_timestamp(stoll(data["ts"].GetString()) * 1e6);
    currency_transfer_bundle_found = true;
  }

  if (!is_transfer_withdrawal_initialized_) {
    is_transfer_withdrawal_initialized_ = true;
    LOG(INFO) << "[ParseRestWithdrawal] Withdrawal ignored: " << transfer_withdrawal_set_.size();
    return;
  }

  if (currency_transfer_bundle_found) {
    LOG(INFO) << "[ParseRestWithdrawal] Withdrawal parsed: "
              << currency_transfer_bundle->each_transfer_size();
    result_.set_type(MsgType::TRANSFER);
  }
}

void OkexV5FuturesParser::ParseFundingFee() {
  /*
  {
      "code":"0",
      "data":[
          {
              "bal":"579871.3540597467563431",
              "balChg":"0.8538136837534231",
              "billId":"616180268004995091",
              "ccy":"USDT",
              "execType":"",
              "fee":"0",
              "from":"",
              "instId":"OP-USDT-SWAP",
              "instType":"SWAP",
              "mgnMode":"cross",
              "notes":"",
              "ordId":"",
              "pnl":"0.8538136837534231",
              "posBal":"0",
              "posBalChg":"0",
              "px":"1.4528",
              "subType":"174",
              "sz":"12877",
              "to":"",
              "ts":"1693180823968",
              "type":"8"
          }
      ],
      "msg":""
  }
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[" << __func__ << "] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsOkexErrMessage(document)) {
    LOG(ERROR) << "[" << __func__ << "] Error message\n" << JsonToString(document);
    return;
  }

  if (!document.HasMember("data") || !document["data"].IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Invalid data format\n" << JsonToString(document);
    return;
  }

  std::vector<FundingFeeElement> funding_fees;
  for (const auto& record : document["data"].GetArray()) {
    if (!record.HasMember("instId") || !record.HasMember("pnl") || !record.HasMember("ccy") ||
        !record.HasMember("ts") || !record.HasMember("billId")) {
      LOG(ERROR) << "[" << __func__ << "] Invalid funding fee format:" << JsonToString(record);
      continue;
    }

    const std::string asset = record["ccy"].GetString();
    const std::string symbol = record["instId"].GetString();
    if (!symcache_->has_currency_native(asset) || !symcache_->has_product_native(symbol)) {
      continue;
    }

    auto* currency = symcache_->GetCurrencyByNative(asset);
    auto* product = symcache_->GetProductByNative(symbol);
    FundingFeeElement funding_fee;

    funding_fee.set_id(record["billId"].GetString());
    funding_fee.set_symbol(product->symbol());
    funding_fee.set_funding_fee(-stod(record["pnl"].GetString()));
    funding_fee.set_funding_fee_currency(currency->currency());
    funding_fee.set_timestamp(stoll(record["ts"].GetString()) * 1'000'000);

    funding_fees.emplace_back(std::move(funding_fee));
  }

  if (is_first_funding_fee_) {
    is_first_funding_fee_ = false;

    for (const auto& funding_fee : funding_fees) {
      consumed_funding_fee_ids_.emplace(funding_fee.id());
    }
    return;
  }

  auto funding_fee_bundle = result_.mutable_funding_fee_bundle();
  funding_fee_bundle->Clear();
  bool funding_fee_bundle_found = false;
  for (auto& funding_fee : funding_fees) {
    if (consumed_funding_fee_ids_.count(funding_fee.id()) != 0) {
      continue;
    }
    consumed_funding_fee_ids_.emplace(funding_fee.id());

    *funding_fee_bundle->add_each_funding_fee() = std::move(funding_fee);
    funding_fee_bundle_found = true;
  }

  if (funding_fee_bundle_found) {
    result_.set_type(BaseParser::MsgType::FUNDING_FEE);
  }
}

void OkexV5FuturesParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (IsOrderUpdate(document)) {
    ParseWsFillInfoList(document);
    ParseWsOrderInfoList(document);
  }

  // Ws position does not give full position at one time, disable
  if (IsPositionUpdate(document)) {
    if (!is_account_position_initialized_) {
      result_.set_type(MsgType::IGNORABLE);
      return;
    }
    ParseWsPositionInfoList(document);
  }
}

auto OkexV5FuturesParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  used_in_python_ = false;

  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg.type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::MULTI_SUBMIT_RESPONSE:
      ParseBatchSubmitResponse();
      break;
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      ParseBatchCancelResponse();
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
      ParseRestOpenOrders();
      break;
    case OrderSubTopicId::REST_DONE_ORDERS:
      ParseRestHistOrders();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillOrders();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestAccountBalance();
      break;
    case OrderSubTopicId::REST_ACCOUNT_POSITION:
      ParseRestAccountPosition();
      break;
    case OkexV5FuturesOrderSubTopicId::REST_DEPOSIT:
      ParseRestDeposit();
      break;
    case OkexV5FuturesOrderSubTopicId::REST_WITHDRAWAL:
      ParseRestWithdrawal();
      break;
    case OrderSubTopicId::REST_FUNDING_FEE:
      ParseFundingFee();
      break;
    default:
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      break;
  }

  return result_;
}

bool OkexV5FuturesParser::CheckCrossModeConsistent(const JsonValue& document, const char* field) {
  if (!document.HasMember(field)) {
    return false;
  }

  if (use_cross_mode_) {
    return std::string(document[field].GetString()) == "cross";
  } else {
    return std::string(document[field].GetString()) == "isolate";
  }
  CHECK(false);
  return true;
}

}  // namespace coin2::exchange::okex_v5_futures::order_v5
